package mr

import (
	"fmt"
	"log"
	"net/rpc"
	"hash/fnv"
	// "math/rand"
	"time"
	"os"
	"io/ioutil"
	"sort"
	"bufio"
	"io"
	"strings"
)


var id int
var nReduce int
var doneCh chan bool
//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }



//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}


//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
	// rand.Seed(time.Now().Unix()) //利用当前时间的UNIX时间戳初始化rand包
	// id = rand.Intn(100)
	// log.Printf("new worker id: %d", id)

	//register
	success,reply := CallRegister();
	
	if !success {
		log.Fatal("regist fail. ")
		log.Printf("exit worker.")
		return
	}
	
	log.Print("worker connect success.")
	id = reply.Id
	data := reply.Data
	interval := data.Heart_interval
	nReduce = data.Nreduce
	doneCh = make(chan bool)

	go KeepAlive(interval)
	go AskAndExecute(mapf, reducef)

	<-doneCh

	fmt.Println("close worker.")
	
}

func AskAndExecute(mapf func(string, string) []KeyValue, reducef func(string, []string) string) {
		for {
			_, reply := callTask()
			task := reply.Data
			taskKind := task.Kind
			
			if taskKind == MAP {
				log.Printf("execute map task %s.", task.Resources)
				execMap(mapf, &task)
			} else if taskKind == REDUCE {
				log.Printf("execute reduce task %s.", task.Resources)
				execReduce(reducef, &task)
			} else if taskKind == SHUTDOWN {
				log.Printf("execute shutdown task.")
				doneCh <- true
				break
			} else if taskKind == EMPTY {
				time.Sleep(time.Duration(1) * time.Second)
			} else {
				fmt.Printf("unsupport task type: %d", taskKind)
			}
		}
}

func callTask() (bool, *GetTaskResp) {
	args := GetTaskReq{id}
	reply := GetTaskResp{}
	success := call("Coordinator.GetTask", &args, &reply)
	return success, &reply
}




func execMap(mapf func(string, string) []KeyValue, task *Task) {

	kvMap := []*[]KeyValue{}
	onames := make([]string, 0)
	for i:= 0; i < nReduce; i++ {
		kvMap = append(kvMap, &[]KeyValue{})
		oname := fmt.Sprintf("mr-%d-%d", task.Member, i)
		onames = append(onames, oname)
	}

	// 读取kvs
	filename := task.Resources[0];
	file, err := os.Open(filename)
	defer file.Close()
	if err != nil {
		fmt.Printf("cannot open %v", filename)
	}
	content, err := ioutil.ReadAll(file)
	if err != nil {
		fmt.Printf("cannot read %v", filename)
	}
	kva := mapf(filename, string(content))

	// keyValue 缓存
	for _,kv := range kva {
		// log.Printf("save %v",kv)
		index  := ihash(kv.Key) % nReduce
		bucket := kvMap[index]
		*bucket = append(*bucket, kv)
	}

	// 输出数据落地
	for i, kvs := range kvMap {
		sort.Sort(ByKey(*kvs))
		ofile,_ := os.Create(onames[i])

		for _, kv := range *kvs {
			_, _ = fmt.Fprintf(ofile, "%v %v\n", kv.Key, kv.Value)
		}
		// log.Printf("%s is persist", onames[i])
	}

	fmt.Printf("finish map task %d\n", task.Member)


	// 解耦
	args := CompleteTaskReq{
		Kind: task.Kind,
		Resources: onames,
		Member: task.Member,
		WorkerId: id,
	}
	reply := CompleteTaskResp{}
	call("Coordinator.CompleteTask", &args, &reply)

}


func execReduce(reducef func(string, []string) string, task *Task) {
	kvs := readFiles(task)
	sort.Sort(ByKey(kvs))
	oname := fmt.Sprintf("mr-out-%d", task.Member)
	ofile, _ := os.Create(oname)
	defer ofile.Close()
	for i := 0;i < len(kvs); {
		j := i + 1
		for j < len(kvs) && kvs[j].Key == kvs[i].Key {
			j++
		}
		values := []string{}
		for k := i; k < j; k++ {
			values = append(values, kvs[k].Value)
		}
		output := reducef(kvs[i].Key, values)

		// this is the correct format for each line of Reduce output.
		fmt.Fprintf(ofile, "%v %v\n", kvs[i].Key, output)
		i = j
	}

	//
	fmt.Printf("finish reduce task %d", task.Member)
	req := CompleteTaskReq{
		Kind: task.Kind,
		Resources: []string{oname},
		Member: task.Member,
		WorkerId: id,
	}
	resp := CompleteTaskResp{}
	call("Coordinator.CompleteTask", &req, &resp)
}

func KeepAlive(interval int) {
	// log.Printf("interval: %d", interval)
	for {
		time.Sleep(time.Duration(interval) * time.Second)
		args := KeepAliveArgs{Id: id}
		var reply KeepAliveReply
		call("Coordinator.Alive", &args, &reply)
		if reply.Success == 1 {
			fmt.Println("keep alive fail.")
			doneCh <- true
		}
		// log.Println("send heart beas.");
		//todo 调用心跳接口失败了怎么办
	}
}

func CallRegister() (bool, *RegisterReply) {
	args := RegisterArgs{}
	reply := RegisterReply{}

	success:= call("Coordinator.Register", &args, &reply)

	return success, &reply
}



//
// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Coordinator.Example", &args, &reply)

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

// 向协调者发送rpc请求，等待响应
// send an RPC request to the coordinator, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := coordinatorSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		log.Fatal("dialing:", err)
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	doneCh <- true

	return false
}


func readFiles(task *Task) []KeyValue {
	res := make([]KeyValue, 0)
    for _, filename := range task.Resources {
		file,_ := os.Open(filename)
		defer file.Close()
		br := bufio.NewReader(file)
		for {
			line, _, c := br.ReadLine()
			if c == io.EOF {
				break
			}
			data := strings.Split(string(line), " ")
			res = append(res, KeyValue {
				Key: data[0],
				Value: data[1],
			})
		}
	}
	return res
}